//===----------------------------------------------------------------------===//
//
//                         BusTub
//
// seq_scan_executor.cpp
//
// Identification: src/execution/seq_scan_executor.cpp
//
// Copyright (c) 2015-2021, Carnegie Mellon University Database Group
//
//===----------------------------------------------------------------------===//

/*
 * 这个执行器是要我们顺序遍历一个Table，并按要求的格式返回遍历结果，如果有过滤条件，也要进行过滤。相当于
 * Select somecols From Test_Table Where ...
 * */

/*
 * seqScanExector遍历表并依次返回一条记录。顺序扫描由seqScanplanNode指定
 * plan node指定被扫描的表，还包含一个predicate，如果该记录不满足predicate，则扫描不会生成记录
 *
 * */
#include "execution/executors/seq_scan_executor.h"

namespace bustub {

// exec_ctx
SeqScanExecutor::SeqScanExecutor(ExecutorContext *exec_ctx, const SeqScanPlanNode *plan) : AbstractExecutor(exec_ctx)
                                                                                           ,plan_(plan)
                                                                                           ,table_heap_(nullptr)
                                                                                            ,iter_(nullptr,RID(), nullptr){}

void SeqScanExecutor::Init() {
  // 对初始信息进行更新
  table_heap_ = exec_ctx_->GetCatalog()->GetTable(plan_->GetTableOid())->table_.get();
  iter_ = table_heap_->Begin(exec_ctx_->GetTransaction());
}


bool SeqScanExecutor::Next(Tuple *tuple, RID *rid) {
  // 遍历完了返回false
  if(iter_==table_heap_->End()){
    return false;
  }
  // 得到RID和要返回的列
//  Tuple(std::vector<Value> values, const Schema *schema);

  RID origin_id = iter_->GetRid();
  const Schema *schema = plan_->OutputSchema();

  // 加锁
  LockManager* lockmanager = GetExecutorContext()->GetLockManager();
  Transaction *txn = GetExecutorContext()->GetTransaction();
  if(lockmanager != nullptr){
    if(txn->GetIsolationLevel() != IsolationLevel::READ_UNCOMMITTED){
      if(!txn->IsSharedLocked(origin_id) && !txn->IsExclusiveLocked(origin_id)){
        lockmanager->LockShared(txn,origin_id);
      }
    }
  }


  // 筛选哪些列要被返回
  std::vector<Value> values;
  values.reserve(schema->GetColumnCount());
  for(size_t i=0;i<values.capacity();++i){
    values.push_back(schema->GetColumn(i).GetExpr()->Evaluate(
        &(*iter_),&(exec_ctx_->GetCatalog()->GetTable(plan_->GetTableOid())->schema_)
        ));
  }

  ++iter_;
//  构造要返回的行
  Tuple res(values,schema);

  // 看看该行符不符合条件，符合则返回，不符合则继续找下一行
  const AbstractExpression *predict = plan_->GetPredicate();
  if(predict== nullptr || predict->Evaluate(&res,schema).GetAs<bool>()){
    *tuple = res;
    *rid = origin_id;
    return true;
  }
  return Next(tuple,rid);



}


}  // namespace bustub
